package reactorexamples;

import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.ArrayList;
import java.util.Random;

/**
 * @author dushougudu
 */
public class FluxDemo {
    public static void main(String[] args) {
        Flux.range(1, 10)
                .timeout(Flux.never(), v -> Flux.never())
                .subscribe(System.out::println);

        Flux.range(1, 4)
                .subscribe(System.out::println,
                        error -> System.err.println("发生错误：" + error),
                        () -> System.out.println("完成"),
                        sub -> {
                            System.out.println("已订阅");
                            // 理解背压
                            // 尝试修改下 request 中的值，看看有啥变化
                            sub.request(10);
                        });

        Flux.create((t) -> {
            t.next("create");
            t.next("create1");
            t.complete();
        }).subscribe(System.out::println);

        Flux.generate(t -> {
            t.next("generate");
            //注意generate中next只能调用1次
            t.complete();
        }).subscribe(System.out::println);

        final Random random = new Random();
        Flux.generate(ArrayList::new, (list, sink) -> {
            int value = random.nextInt(100);
            list.add(value);
            sink.next(value);
            if (list.size() == 10) {
                sink.complete();
            }
            return list;
        }).subscribe(System.out::println);

        Flux f1 = Flux.just(System.currentTimeMillis());
        f1.subscribe(System.out::println);
        f1.subscribe(System.out::println);

        Flux f2 = Flux.defer(() -> Flux.just(System.currentTimeMillis()));
        f2.subscribe(System.out::println);
        f2.subscribe(System.out::println);

        Flux.merge(Flux.interval(Duration.of(0, ChronoUnit.MILLIS), Duration.of(100, ChronoUnit.MILLIS)).take(5),
                Flux.interval(Duration.of(0, ChronoUnit.MILLIS), Duration.of(100, ChronoUnit.MILLIS)).take(5))
                .toStream()
                .forEach(System.out::println);

        Flux.mergeSequential(Flux.interval(Duration.of(0, ChronoUnit.MILLIS), Duration.of(100, ChronoUnit.MILLIS)).take(5),
                Flux.interval(Duration.of(50, ChronoUnit.MILLIS), Duration.of(100, ChronoUnit.MILLIS)).take(5))
                .toStream()
                .forEach(System.out::println);


        Flux.empty().subscribe(System.out::println, System.out::println, () -> System.out.println("结束"));

        Flux.never().subscribe(System.out::println, System.out::println, () -> System.out.println("结束"));

    }
}
